package io.github.kuafucv.zero.log;

import ch.qos.logback.classic.spi.ILoggingEvent;
import io.github.kuafucv.zero.log.context.ZeroLogContext;
import io.github.kuafucv.zero.log.http.HttpBuilder;
import io.github.kuafucv.zero.log.io.DiscardingRollingOutput;
import io.github.kuafucv.zero.log.io.IoUtils;
import java.io.*;
import java.net.HttpURLConnection;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class HttpBatchAppender extends AbstractHttpAppender{

    private int flushIntervalInSeconds = 3;
    public void setFlushIntervalInSeconds(int flushIntervalInSeconds) {
        this.flushIntervalInSeconds = flushIntervalInSeconds;
    }

    private int maxNumberOfBuckets = 8;
    public void setMaxNumberOfBuckets(int maxNumberOfBuckets) {
        this.maxNumberOfBuckets = maxNumberOfBuckets;
    }
    /**
     * 存储桶大小（KB）
     */
    private int maxBucketSizeInKilobytes = 1024;
    public void setMaxBucketSizeInKilobytes(int maxBucketSizeInKilobytes) {
        this.maxBucketSizeInKilobytes = maxBucketSizeInKilobytes;
    }

    private DiscardingRollingOutput outputStream;
    protected final AtomicLong sendDurationInNanos = new AtomicLong();
    protected final AtomicLong sentBytes = new AtomicLong();
    protected final AtomicInteger sendSuccessCount = new AtomicInteger();
    protected final AtomicInteger sendExceptionCount = new AtomicInteger();
    private ScheduledExecutorService scheduledExecutor;

    @Override
    protected void append(ILoggingEvent eventObject) {
        if(isStarted()){
            String msg = null;
            try {
                msg = this.convertEventToJson(eventObject);
            } catch (Exception e) {
                addError("Failed format log to json", e);
            }
            if(msg != null && !msg.isEmpty()){
                ZeroLogContext.offerJsonLog(msg);
            }
        }
    }

    @Override
    public void start() {
        outputStream = new DiscardingRollingOutput(maxBucketSizeInKilobytes * 1024, maxNumberOfBuckets);
        // 调度
        ThreadFactory threadFactory = new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = Executors.defaultThreadFactory().newThread(r);
                thread.setName("zero-log-http-batch-appender");
                thread.setDaemon(true);
                return thread;
            }
        };
        scheduledExecutor = Executors.newSingleThreadScheduledExecutor(threadFactory);
        scheduledExecutor.scheduleWithFixedDelay(new LogExporter(), flushIntervalInSeconds, flushIntervalInSeconds, TimeUnit.SECONDS);
        JsonLogQueueConsumer jsonLogQueueConsumer = new JsonLogQueueConsumer();
        Thread jsonLogQueueConsumerThread = new Thread(jsonLogQueueConsumer);
        jsonLogQueueConsumerThread.setName("json-log-queue-consumer");
        jsonLogQueueConsumerThread.setDaemon(true);
        jsonLogQueueConsumerThread.start();
        super.start();
    }

    @Override
    public void stop() {
        scheduledExecutor.shutdown();
        processLogEntries();
        try {
            scheduledExecutor.awaitTermination(2L * this.flushIntervalInSeconds, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            addWarn("Exception waiting for termination of HttpBatchAppender scheduler", e);
        }
        outputStream.close();
        super.stop();
    }

    public class JsonLogQueueConsumer implements Runnable {
        @Override
        public void run() {
            while (true) {
                String take;
                try {
                    take = ZeroLogContext.takeJsonLog();
                } catch (InterruptedException e) {
                    break;
                }
                try {
                    outputStream.write(take.getBytes(StandardCharsets.UTF_8));
                } catch (Exception e) {
                    addError("Failed json log to outputStream", e);
                }
            }
        }
    }

    public class LogExporter implements Runnable {
        @Override
        public void run() {
            try {
                processLogEntries();
            } catch (Exception e) {
                addWarn("Internal error", e);
            }
        }
    }

    private void processLogEntries() {
        try {
            outputStream.rollCurrentBucketIfNotEmpty();
        }catch (IOException e){
            addWarn("Internal error", e);
        }
        BlockingDeque<ByteArrayOutputStream> filledBuckets = outputStream.getFilledBuckets();

        ByteArrayOutputStream bucket;

        while ((bucket = filledBuckets.poll()) != null) {
            if(isStarted()){
                try {
                    InputStream in = new ByteArrayInputStream(bucket.toByteArray());
                    processLogEntries(in);
                    in.close();
                } catch (Exception e) {
                    addWarn("Internal error", e);
                }
            }
            outputStream.recycleBucket(bucket);
        }
    }

    private void processLogEntries(InputStream in) {
        long nanosBefore = System.nanoTime();
        try {
            HttpURLConnection conn = HttpBuilder.build();
            BufferedOutputStream out = new BufferedOutputStream(conn.getOutputStream());
            long len = IoUtils.copy(in, out);
            sentBytes.addAndGet(len);
            out.flush();
            out.close();
            int responseCode = conn.getResponseCode();
            String response = this.readResponseBody(conn.getInputStream());
            switch (responseCode) {
                case HttpURLConnection.HTTP_OK:
                case HttpURLConnection.HTTP_ACCEPTED:
                    sendSuccessCount.incrementAndGet();
                    break;
                default:
                    sendExceptionCount.incrementAndGet();
                    addWarn("Log server exception: " + responseCode + ": " + response);
            }
            try {
                conn.getInputStream().close();
                conn.disconnect();
            } catch (Exception e) {
                addWarn("HttpURLConnection close or disconnect exception", e);
            }
        } catch (Exception e) {
            sendExceptionCount.incrementAndGet();
            addWarn("ProcessLogEntries exception", e);
        } finally {
            sendDurationInNanos.addAndGet(System.nanoTime() - nanosBefore);
        }
    }

    private String readResponseBody(final InputStream input) throws IOException {
        try {
            final byte[] bytes = toBytes(input);
            return new String(bytes, StandardCharsets.UTF_8);
        } finally {
            input.close();
        }
    }


    public long getSentBytes() {
        return sentBytes.get();
    }

    public long getSendDurationInNanos() {
        return sendDurationInNanos.get();
    }

    public int getSendSuccessCount() {
        return sendSuccessCount.get();
    }

    public int getSendExceptionCount() {
        return sendExceptionCount.get();
    }

    public int getDiscardedBucketsCount() {
        return outputStream.getDiscardedBucketCount();
    }

    public long getCurrentLogEntriesBufferSizeInBytes() {
        return outputStream.getCurrentOutputStreamSize();
    }

    private String getDebugInfo() {
        return "{" +
                "sendDurationInMillis=" + TimeUnit.MILLISECONDS.convert(sendDurationInNanos.get(), TimeUnit.NANOSECONDS) +
                ", sendSuccessCount=" + sendSuccessCount +
                ", sendExceptionCount=" + sendExceptionCount +
                ", sentBytes=" + sentBytes +
                ", discardedBucketsCount=" + getDiscardedBucketsCount() +
                ", currentLogEntriesBufferSizeInBytes=" + getCurrentLogEntriesBufferSizeInBytes() +
                '}';
    }

}
